package com.mq.produce.simple;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.HashMap;
import java.util.concurrent.Future;

/**
 * @Classname KafkaSimpleProduce
 * @Description
 * @Version 1.0.0
 * @Date 2024/12/4 15:17
 * @Created by ncs
 */
public class KafkaSimpleProduce {

    public static void main(String[] args) throws Exception{
        //1. 创建生产者对象  需要范型 k,v
        // 1.1 配置对象
        HashMap<String, Object> configMap = new HashMap<>();
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.125.100:9093");
        // 1.2 序列化
        configMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        configMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //自定义拦截器
        configMap.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName());
        //自定义分区器
        configMap.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName());
        //broker应答级别 0:leader接受成功 1:leader落盘 -1或者all:同步给了isr列表里面的所有副本 默认是-1
        configMap.put(ProducerConfig.ACKS_CONFIG, "-1");
        //没有收到预期的acks应答 那么重试五次发送消息 默认是int的最大值
        configMap.put(ProducerConfig.RETRIES_CONFIG, 5);
        //默认16k 单位字节 批次数据 消息头65+消息体(动态)
        configMap.put(ProducerConfig.BATCH_SIZE_CONFIG, 5);
        //超时时间 默认30s
        configMap.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 3000);
        //生产者幂等配置 能够保证acks应答出问题的时候 消息不乱序和消息不重复
        configMap.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        //增加事务id  为了解决生产者断开连接以后再发送消息 能够保证幂等操作有效
        configMap.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-tx-id");
        KafkaProducer<String, String> kafkaProduce = new KafkaProducer<>(configMap);
        //初始化事务
        kafkaProduce.initTransactions();
        //2. 创建数据
        //2.1 第一个参数表示主题名称  第二个参数表示数据的KEY 第三个参数表示数据的VALUE
        kafkaProduce.beginTransaction();
        try {
            for (int i = 0; i < 100; i++) {
                ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                        "czp_topic","key" + i, "value" + i
                );
//            kafkaProduce.send(producerRecord, (metadata, exception) -> System.out.println("callback发送数据成功，收到了应答" + metadata));
//            System.out.println("消息发送成功key=" + "key" + i + "    value=" + "value" + i);

                Future<RecordMetadata> send = kafkaProduce.send(producerRecord);
                RecordMetadata recordMetadata = send.get();
                System.out.println("callback发送数据成功，收到了应答" + recordMetadata);
            }
        }catch (Exception e){
            kafkaProduce.abortTransaction();
        }
        kafkaProduce.commitTransaction();
        // 4. 关闭连接
        kafkaProduce.close();
    }
}
